package com.o2o.cleaning.month.platform.ebusiness_plat.kuaishou.shuang11

import org.apache.spark.sql.SparkSession

object _05_DWS_TO_DWS {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName(s"${this.getClass.getSimpleName}")
      .config("spark.debug.maxToStringFields", "500")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.caseSensitive", "true")
      .config("spark.executor.heartbeatInterval", "120s")
      .config("spark.network.timeout", "12000s")
      //      .config("spark.default.parallelism", "2")
      .config("cluster.name", "O2OElastic")
      .config("spark.sql.crossJoin.enabled", "true")
      .master("local[*]")
      //      .enableHiveSupport()
      .getOrCreate()

    val sc = spark.sparkContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")
    sc.setLogLevel("WARN")

    /**
      * 这个是外表，跑完直接进去
      */
    val year = 2022
    val month = 1111
    val timestamp = 1668096000
    val platformName = "kuaishou"
    val datawork_path = s"s3a://o2o-tempdata/zyf/livestreaming/kuaishou/dws/${year}/${month}/"
    val result_path = s"s3a://o2o-tempdata/zyf/livestreaming/kuaishou/es/${year}/${month}/"
    spark.read.orc(datawork_path).registerTempTable("t_source1")
    //    spark.read.orc(datawork_path).printSchema()
    spark.read.orc(result_path).registerTempTable("t_source2")
    //    spark.read.orc(result_path).printSchema()
    spark.sql(
      s"""
         |select
         |t1.good_id good_id,
         |t1.platform_id platform_id,
         |t1.title title,
         |t1.images images,
         |t1.sellcount sellcount,
         |t1.salesamount salesamount,
         |t1.final_price final_price,
         |t1.final_price_int final_price_int,
         |t1.price price,
         |t1.price_int price_int,
         |t1.original_cost original_cost,
         |t1.coupon_list coupon_list,
         |t1.price_abnormal_degree price_abnormal_degree,
         |t1.quantity quantity,
         |t1.fans fans,
         |t1.allitemcount allitemcount,
         |t1.evaluates evaluates,
         |t1.creditlevel creditlevel,
         |t1.goodratepercentage goodratepercentage,
         |t1.favcount favcount,
         |t1.commentcount commentcount,
         |t1.is_newproduct is_newproduct,
         |t1.is_onsell is_onsell,
         |t1.is_oldfornew is_oldfornew,
         |t1.is_jhs is_jhs,
         |t1.goodurl goodurl,
         |t1.base_info base_info,
         |t2.originProduction originProduction,
         |t2.originItem originItem,
         |t2.originCountryName originCountryName,
         |t2.taxFreeZone taxFreeZone,
         |t1.bc_type bc_type,
         |t1.is_oversea is_oversea,
         |t1.is_market is_market,
         |t1.shopid shopid,
         |t1.userid userid,
         |t1.shopname shopname,
         |t1.shopurl shopurl,
         |t1.starts starts,
         |t1.flagship_type flagship_type,
         |t1.is_flagship is_flagship,
         |t1.is_selfsupport is_selfsupport,
         |t2.name company_name,
         |t2.regional_ID region_id,
         |t2.province province,
         |t2.city city,
         |t2.district district,
         |t1.brand_id brand_id,
         |t1.brand_name brand_name,
         |t2.brandValueId standard_brand_id,
         |t2.brandName_cn standard_brand_name,
         |t2.brand_state brand_state,
         |t2.brand_type brand_type,
         |case when t2.brand_isLaoZiHao = '0' then false else true end is_laozihao,
         |t1.laozihao_type laozihao_type,
         |t2.parentbrandid parentbrandid,
         |t1.lastcategoryid lastcategoryid,
         |t1.lastcategoryname lastcategoryname,
         |t1.rootcategoryid rootcategoryid,
         |t1.rootcategoryname rootcategoryname,
         |t1.categoryid categoryid,
         |t1.categoryname categoryname,
         |t1.subcategoryid subcategoryid,
         |t1.subcategoryname subcategoryname,
         |t1.detailcategoryid detailcategoryid,
         |t1.detailcategoryname detailcategoryname,
         |case when t2.fourthCategoryId is not null then t2.fourthCategoryId else case when t2.thirdCategoryId is not null then t2.thirdCategoryId else case when t2.secondCategoryId is not null then t2.secondCategoryId else case when t2.firstCategoryId is not null then t2.firstCategoryId else '-1' end end end end standardlastcategoryid,
         |t2.firstCategoryId firstcategoryid,
         |t2.secondCategoryId secondcategoryid,
         |t2.thirdCategoryId thirdcategoryid,
         |t2.fourthCategoryId fourthcategoryid,
         |'' fifthcategoryid,
         |${timestamp} timestamp,
         |t1.random_int_val random_int_val,
         |t2.brand_company_name brand_company_name,
         |t1.is_live_good is_live_good,
         |t1.live_count live_count,
         |t1.anchor_count anchor_count,
         |t1.visitor_count visitor_count,
         |t1.shop_live_count shop_live_count,
         |t1.shopimages shopimages,
         |t2.address address
         |from t_source1 t1
         |left join(select good_id,
         |          max(originProduction) originProduction,
         |          max(originItem) originItem,
         |          max(originCountryName) originCountryName,
         |          max(taxFreeZone) taxFreeZone,
         |          max(name) name,
         |          max(regional_ID) regional_ID,
         |          max(province) province,
         |          max(city) city,
         |          max(district) district,
         |          max(brandValueId) brandValueId,
         |          max(brandName_cn) brandName_cn,
         |          max(brand_state) brand_state,
         |          max(brand_type) brand_type,
         |          max(brand_isLaoZiHao) brand_isLaoZiHao,
         |          max(parentbrandid) parentbrandid,
         |          max(firstCategoryId) firstCategoryId,
         |          max(secondCategoryId) secondCategoryId,
         |          max(thirdCategoryId) thirdCategoryId,
         |          max(fourthCategoryId) fourthCategoryId,
         |          max(brand_company_name) brand_company_name,
         |          max(address) address
         |          from t_source2 group by good_id) t2
         |on t1.good_id = t2.good_id
         |""".stripMargin).coalesce(30).write.mode("overwrite").orc(s"s3a://o2o-tempdata/zyf/livestreaming/kuaishou/dws_to_dws/${year}/${month}/")
  }
}
